package com.haoziqi.chapter_06;

import com.alibaba.fastjson.JSON;
import com.haoziqi.chapter_02.WaterSensor;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import org.apache.kafka.clients.producer.KafkaProducer;

import java.util.ArrayList;
import java.util.Properties;

/**
 * description
 * created by A on 2021/3/5
 */
public class KafkaSink_Demo {
    public static void main(String[] args) {

        //1.创建环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStreamSource<String> inputDS = env.socketTextStream("localhost", 9999);
        Properties properties=new Properties();
        properties.setProperty("bootstrap.servers", "hadoop1:9092,hadoop2:9092,hadoop3:9092");
        FlinkKafkaProducer<String> flink0923 = new FlinkKafkaProducer<>("flink0923", new SimpleStringSchema(), properties);
        inputDS.addSink(flink0923);
        //提交任务
        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }



    }
}
